
package com.yimixian.app.rest;

import com.baidu.mapapi.model.LatLng;
import com.google.gson.FieldNamingPolicy;
import com.google.gson.Gson;
import com.google.gson.GsonBuilder;
import com.squareup.okhttp.OkHttpClient;
import com.yimixian.app.BuildConfig;
import com.yimixian.app.R;
import com.yimixian.app.SystemFramework;
import com.yimixian.app.address.MutableAddress;
import com.yimixian.app.common.CacheKeys;
import com.yimixian.app.common.Constants;
import com.yimixian.app.common.Lists;
import com.yimixian.app.common.SharePreferenceKeys;
import com.yimixian.app.data.DataManager;
import com.yimixian.app.model.Address;
import com.yimixian.app.model.AdvertInfo;
import com.yimixian.app.model.AppConfig;
import com.yimixian.app.model.KV;
import com.yimixian.app.model.Order;
import com.yimixian.app.model.PayWithQQWalletResponse;
import com.yimixian.app.model.PullMessage;
import com.yimixian.app.model.QqwalletPayOrder;
import com.yimixian.app.model.Store;
import com.yimixian.app.model.User;
import com.yimixian.app.model.ValidatedOrder;
import com.yimixian.app.model.WechatOrder;
import com.yimixian.app.net.NetConstants;
import com.yimixian.app.rest.response.AppConfigResponse;
import com.yimixian.app.rest.response.AsyncCreateOrderResponse;
import com.yimixian.app.rest.response.BaseResponse;
import com.yimixian.app.rest.response.CheckUserResponse;
import com.yimixian.app.rest.response.CreateAddressResponse;
import com.yimixian.app.rest.response.CreateOrderResponse;
import com.yimixian.app.rest.response.GetAdvertInfoResponse;
import com.yimixian.app.rest.response.GetStoreByLocationResponse;
import com.yimixian.app.rest.response.GetStoreResponse;
import com.yimixian.app.rest.response.PayWithAlipayResponse;
import com.yimixian.app.rest.response.PayWithWechatResponse;
import com.yimixian.app.rest.response.PullMessageResponse;
import com.yimixian.app.rest.response.ShowAddressesResponse;
import com.yimixian.app.rest.response.ShowCurrentUserResponse;
import com.yimixian.app.rest.response.StoresResponse;
import com.yimixian.app.rest.response.UpdateUserInfoResponse;
import com.yimixian.app.rest.response.ValidateOrderResponse;
import com.yimixian.app.util.SharedPreferencesHelper;

import retrofit.RequestInterceptor;
import retrofit.RestAdapter;
import retrofit.client.OkClient;
import retrofit.converter.GsonConverter;
import rx.Observable;
import rx.Subscriber;
import rx.schedulers.Schedulers;

import android.util.Log;
import android.util.Pair;

import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.TimeUnit;

public class YmxDataService {

    public static final String RELEASE_ENDPOINT = "http://api.1mxian.com/";
    public static final String STAGING_ENDPOINT = "http://api.staging.1mxian.com/";
    public static final String TEST_ENDPOINT = "http://api.test.1mxian.com/";
    public static final String MOCK_ENDPOINT = "http://api.mock.1mxian.com/";
    public static final String RD_ENDPOINT = "http://api.rd.1mxian.com/";

    private static final long CONNECT_TIMEOUT_MILLIS = 20 * 1000;
    private static final long READ_TIMEOUT_MILLIS = 30 * 1000;
    private static final String TAG = "retrofit";

    public static YmxDataService sInstance;

    public static YmxDataService getInstance() {
        if (sInstance == null) {
            sInstance = new YmxDataService(DataManager.getInstance());
        }
        return sInstance;
    }

    private final DataManager mDataManager;
    private final YmxRestService mYmxRestService;
    private final String mReferFrom;

    public YmxDataService(DataManager dataManager) {
        mDataManager = dataManager;
        mReferFrom = "APP_ANDROID_" + mDataManager.get(CacheKeys.VERSION_NAME) + "_" +
                mDataManager.get(CacheKeys.INSTALL_CHANNEL);

        Gson gson = new GsonBuilder()
                .setFieldNamingPolicy(FieldNamingPolicy.LOWER_CASE_WITH_UNDERSCORES)
                .setDateFormat("yyyy-MM-dd HH:mm:ss")
                .create();
        GsonConverter gsonConverter = new GsonConverter(gson);
        OkHttpClient client = new OkHttpClient();
        client.setConnectTimeout(CONNECT_TIMEOUT_MILLIS, TimeUnit.MILLISECONDS);
        client.setReadTimeout(READ_TIMEOUT_MILLIS, TimeUnit.MILLISECONDS);

        final String userAgent = "Android YMXian/" + mDataManager.get(CacheKeys.VERSION_NAME);
        RequestInterceptor interceptor = new RequestInterceptor() {
            @Override
            public void intercept(RequestFacade request) {
                request.addHeader("User-Agent", userAgent);
            }
        };
        RestAdapter.Builder builder =
                new RestAdapter.Builder()
                        .setEndpoint(getEndPoint())
                        .setConverter(gsonConverter)
                        .setRequestInterceptor(interceptor)
                        .setClient(new OkClient(client));
        if (BuildConfig.DEBUG) {
            builder.setLogLevel(RestAdapter.LogLevel.FULL)
                    .setLog(new RestAdapter.Log() {
                        @Override
                        public void log(String message) {
                            Log.e(TAG, message);
                        }
                    });
        }
        RestAdapter restAdapter = builder.build();
        mYmxRestService = restAdapter.create(YmxRestService.class);
        String host = getEndPoint().substring(7);
        host = host.substring(0, host.length() - 1);
        NetConstants.HTTP_HOST = host;
    }

    public Observable<AppConfig> getAppConfig(final boolean readFromCache) {
        return Observable.create(new Observable.OnSubscribe<AppConfig>() {
            @Override
            public void call(Subscriber<? super AppConfig> subscriber) {
                if (subscriber.isUnsubscribed()) {
                    return;
                }

                if (readFromCache) {
                    AppConfig appConfig = mDataManager.get(CacheKeys.APP_CONFIG);
                    if (appConfig != null) {
                        SharedPreferencesHelper.commitString(Constants.CONNECT_SERVICE,
                                appConfig.customerServiceTel);
                        subscriber.onNext(appConfig);
                        subscriber.onCompleted();
                        return;
                    }
                }

                AppConfigResponse response = mYmxRestService.getAppConfig(mDataManager
                        .<String> get(CacheKeys.TOKEN));
                if (response.hasError()) {
                    throw new RuntimeException(response.error.message);
                }

                mDataManager.put(CacheKeys.APP_CONFIG, response.appConfig);
                SharedPreferencesHelper.commitString(Constants.CONNECT_SERVICE,
                        response.appConfig.customerServiceTel);
              if(response != null && response.appConfig != null) {
                  SharedPreferencesHelper.saveAppConfig(response.appConfig);
              }

                subscriber.onNext(response.appConfig);
                subscriber.onCompleted();
            }
        }).subscribeOn(Schedulers.io());
    }

    public Observable<User> getCurrentUser(final boolean readFromCache) {
        return Observable.create(new Observable.OnSubscribe<User>() {
            @Override
            public void call(Subscriber<? super User> subscriber) {
                if (subscriber.isUnsubscribed()) {
                    return;
                }

                String token = mDataManager.get(CacheKeys.TOKEN);
                if (token == null) {
                    return;
//                    throw new RuntimeException(SystemFramework.getInstance().getGlobalContext()
//                            .getString(R.string.user_unlogin));
                }

                if (readFromCache) {
                    User user = mDataManager.get(CacheKeys.CURRENT_USER);
                    if (user != null) {
                        subscriber.onNext(user);
                        subscriber.onCompleted();
                        return;
                    }
                }

                ShowCurrentUserResponse response = mYmxRestService.getCurrentUser(token);
                if (response.hasError()) {
                    mDataManager.remove(CacheKeys.TOKEN);
                    SharedPreferencesHelper.clearUser(SharePreferenceKeys.CURRENT_USER);
                    SharedPreferencesHelper.clearAddress(SharePreferenceKeys.ADDRESSES);
                    SharedPreferencesHelper.clearAddress(SharePreferenceKeys.CURRENT_ADDRESS);
                    throw new RuntimeException(response.error.message);
                }
                User user = response.user;
                user.token = token;
                mDataManager.put(CacheKeys.CURRENT_USER, user);
                SharedPreferencesHelper.saveUser(response.user, SharePreferenceKeys.CURRENT_USER);
                subscriber.onNext(response.user);
                subscriber.onCompleted();
            }
        }).subscribeOn(Schedulers.io());
    }

    public Observable<List<Address>> getAddresses(final boolean readFromCache) {
        return Observable.create(new Observable.OnSubscribe<List<Address>>() {
            @Override
            public void call(Subscriber<? super List<Address>> subscriber) {
                if (subscriber.isUnsubscribed()) {
                    return;
                }
                String token = DataManager.getInstance().get(CacheKeys.TOKEN);
                if (token == null) {
                    throw new RuntimeException(SystemFramework.getInstance().getGlobalContext()
                            .getString(R.string.user_unlogin));
                }
                if (readFromCache) {
                    List<Address> addresses = SharedPreferencesHelper
                            .getAddresses(SharePreferenceKeys.ADDRESSES);
                    if (addresses != null && addresses.size() != 0) {
                        subscriber.onNext(addresses);
                        subscriber.onCompleted();
                        return;
                    }
                }
                ShowAddressesResponse response = mYmxRestService.getAddresses(token,
                        Constants.BAIDU_COORD_SYSTEM);
                if (response.hasError()) {
                    throw new RuntimeException(response.error.message);
                }
                SharedPreferencesHelper.saveAddresses((ArrayList) response.addresses,
                        SharePreferenceKeys.ADDRESSES);
                subscriber.onNext(Lists.nullToEmpty(response.addresses));
                subscriber.onCompleted();
            }
        }).subscribeOn(Schedulers.io());
    }

    public Observable<Address> createAddress(final MutableAddress address) {
        return Observable.create(new Observable.OnSubscribe<Address>() {
            @Override
            public void call(Subscriber<? super Address> subscriber) {
                if (subscriber.isUnsubscribed()) {
                    return;
                }

                String token = mDataManager.get(CacheKeys.TOKEN);
                if (token == null) {
                    throw new RuntimeException(SystemFramework.getInstance().getGlobalContext()
                            .getString(R.string.user_unlogin));
                }

                CreateAddressResponse response = mYmxRestService.createAddress(
                        token,
                        address.name,
                        address.gender,
                        address.tel,
                        address.province,
                        address.provinceKey,
                        address.city,
                        address.cityKey,
                        address.district,
                        address.districtKey,
                        address.poiId,
                        address.poiName,
                        address.poiAddress,
                        address.roomNo,
                        address.tag,
                        String.valueOf(address.longitude),
                        String.valueOf(address.latitude),
                        Constants.BAIDU_COORD_SYSTEM);
                if (response.hasError()) {
                    throw new RuntimeException(response.error.message);
                }
                subscriber.onNext(response.address);
                subscriber.onCompleted();
            }
        }).subscribeOn(Schedulers.io());
    }

    public Observable<Boolean> updateAddress(final MutableAddress address) {
        return Observable.create(new Observable.OnSubscribe<Boolean>() {
            @Override
            public void call(Subscriber<? super Boolean> subscriber) {
                if (subscriber.isUnsubscribed()) {
                    return;
                }

                String token = mDataManager.get(CacheKeys.TOKEN);
                if (token == null) {
                    throw new RuntimeException(SystemFramework.getInstance().getGlobalContext()
                            .getString(R.string.user_unlogin));
                }

                BaseResponse response = mYmxRestService.updateAddress(
                        String.valueOf(address.id),
                        token,
                        address.name,
                        address.gender,
                        address.tel,
                        address.province,
                        address.provinceKey,
                        address.city,
                        address.cityKey,
                        address.district,
                        address.districtKey,
                        address.poiId,
                        address.poiName,
                        address.poiAddress,
                        address.roomNo,
                        address.tag,
                        String.valueOf(address.longitude),
                        String.valueOf(address.latitude),
                        Constants.BAIDU_COORD_SYSTEM);
                if (response.hasError()) {
                    throw new RuntimeException(response.error.message);
                }
                subscriber.onNext(true);
                subscriber.onCompleted();
            }
        }).subscribeOn(Schedulers.io());
    }

    public Observable<Boolean> deleteAddress(final int addressId) {
        return Observable.create(new Observable.OnSubscribe<Boolean>() {
            @Override
            public void call(Subscriber<? super Boolean> subscriber) {
                if (subscriber.isUnsubscribed()) {
                    return;
                }

                String token = mDataManager.get(CacheKeys.TOKEN);
                if (token == null) {
                    throw new RuntimeException(SystemFramework.getInstance().getGlobalContext()
                            .getString(R.string.user_unlogin));
                }

                BaseResponse response = mYmxRestService.deleteAddress(String.valueOf(addressId),
                        token);
                if (response.hasError()) {
                    throw new RuntimeException(response.error.message);
                }

                subscriber.onNext(true);
                subscriber.onCompleted();
            }
        }).subscribeOn(Schedulers.io());
    }

    public Observable<Store> getStoreByAddressId(final String addressId, final String token) {
        return Observable.create(new Observable.OnSubscribe<Store>() {
            @Override
            public void call(Subscriber<? super Store> subscriber) {
                if (subscriber.isUnsubscribed()) {
                    return;
                }

                GetStoreByLocationResponse response = mYmxRestService.getStoreByLocation(
                        addressId,
                        null,
                        null,
                        null,
                        null,
                        token);
                if (response.hasError()) {
                    throw new RuntimeException(response.error.message);
                }

                subscriber.onNext(response.store);
                subscriber.onCompleted();
            }
        }).subscribeOn(Schedulers.io());
    }

    public Observable<Store> getStoreByLatLon(final LatLng latLng, final String poiId,
            final String token) {
        return Observable.create(new Observable.OnSubscribe<Store>() {
            @Override
            public void call(Subscriber<? super Store> subscriber) {
                if (subscriber.isUnsubscribed()) {
                    return;
                }

                GetStoreByLocationResponse response = mYmxRestService.getStoreByLocation(
                        null,
                        poiId,
                        String.valueOf(latLng.longitude),
                        String.valueOf(latLng.latitude),
                        Constants.BAIDU_COORD_SYSTEM,
                        token);
                if (response.hasError()) {
                    throw new RuntimeException(response.error.message);
                }

                subscriber.onNext(response.store);
                subscriber.onCompleted();
            }
        }).subscribeOn(Schedulers.io());
    }

    public Observable<Boolean> registerPushToken(final String clientId, final String token) {
        return Observable.create(new Observable.OnSubscribe<Boolean>() {
            @Override
            public void call(Subscriber<? super Boolean> subscriber) {
                if (subscriber.isUnsubscribed()) {
                    return;
                }

                BaseResponse response = mYmxRestService.registerPushToken(token, "getui", clientId);
                if (response.hasError()) {
                    throw new RuntimeException(response.error.message);
                }

                subscriber.onNext(true);
                subscriber.onCompleted();
            }
        }).subscribeOn(Schedulers.io());
    }

    public Observable<String> payWithAlipay(final String paymentId) {
        return Observable.create(new Observable.OnSubscribe<String>() {
            @Override
            public void call(Subscriber<? super String> subscriber) {
                if (subscriber.isUnsubscribed()) {
                    return;
                }

                PayWithAlipayResponse response = mYmxRestService.payWithAlipay(
                        paymentId,
                        mDataManager.<String>get(CacheKeys.TOKEN));
                if (response.hasError()) {
                    throw new RuntimeException(response.error.message);
                }

                subscriber.onNext(response.alipayOrderString);
                subscriber.onCompleted();
            }
        }).subscribeOn(Schedulers.io());
    }

    public Observable<WechatOrder> payWithWechat(final String paymentId) {
        return Observable.create(new Observable.OnSubscribe<WechatOrder>() {
            @Override
            public void call(Subscriber<? super WechatOrder> subscriber) {
                if (subscriber.isUnsubscribed()) {
                    return;
                }

                PayWithWechatResponse response = mYmxRestService.payWithWechat(
                        paymentId,
                        mDataManager.<String>get(CacheKeys.TOKEN));
                if (response.hasError()) {
                    throw new RuntimeException(response.error.message);
                }

                subscriber.onNext(response.wechatOrder);
                subscriber.onCompleted();
            }
        }).subscribeOn(Schedulers.io());
    }

  public Observable<QqwalletPayOrder> payWithQQWallet(final String paymentId) {
    return Observable.create(new Observable.OnSubscribe<QqwalletPayOrder>() {
      @Override
      public void call(Subscriber<? super QqwalletPayOrder> subscriber) {
        if (subscriber.isUnsubscribed()) {
          return;
        }

        PayWithQQWalletResponse response = mYmxRestService.payWithQQWallet(
                paymentId,
                mDataManager.<String>get(CacheKeys.TOKEN));
        if (response.hasError()) {
          throw new RuntimeException(response.error.message);
        }

        subscriber.onNext(response.qqwalletPayOrder);
        subscriber.onCompleted();
      }
    }).subscribeOn(Schedulers.io());
  }

    public Observable<Boolean> payWithBalance(final String paymentId) {
        return Observable.create(new Observable.OnSubscribe<Boolean>() {
            @Override
            public void call(Subscriber<? super Boolean> subscriber) {
                if (subscriber.isUnsubscribed()) {
                    return;
                }

                BaseResponse response = mYmxRestService.payWithBalance(
                        paymentId,
                        mDataManager.<String>get(CacheKeys.TOKEN));
                if (response.hasError()) {
                    throw new RuntimeException(response.error.message);
                }

                subscriber.onNext(true);
                subscriber.onCompleted();
            }
        }).subscribeOn(Schedulers.io());
    }

    public Observable<Order> createOrderByAddressId(
            final int addressId,
            final int deliveryModeId,
            final String goodsJsonString,
            final LatLng latLng) {
        return Observable.create(new Observable.OnSubscribe<Order>() {
            @Override
            public void call(Subscriber<? super Order> subscriber) {
                if (subscriber.isUnsubscribed()) {
                    return;
                }

                CreateOrderResponse response = mYmxRestService.createOrderByAddressId(
                        mReferFrom,
                        (String) mDataManager.get(CacheKeys.TOKEN),
                        addressId,
                        deliveryModeId,
                        goodsJsonString,
                        latLng != null ? String.valueOf(latLng.longitude) : null,
                        latLng != null ? String.valueOf(latLng.latitude) : null);
                if (response.hasError()) {
                    throw new RuntimeException(response.error.message);
                }

                subscriber.onNext(response.order);
                subscriber.onCompleted();
            }
        }).subscribeOn(Schedulers.io());
    }

    public Observable<Order> createOrderByStoreId(
            final int storeId,
            final int deliveryModeId,
            final String goodsJsonString,
            final LatLng latLng) {
        return Observable.create(new Observable.OnSubscribe<Order>() {
            @Override
            public void call(Subscriber<? super Order> subscriber) {
                if (subscriber.isUnsubscribed()) {
                    return;
                }

                CreateOrderResponse response = mYmxRestService.createOrderByStoreId(
                        mReferFrom,
                        (String) mDataManager.get(CacheKeys.TOKEN),
                        storeId,
                        deliveryModeId,
                        goodsJsonString,
                        latLng != null ? String.valueOf(latLng.longitude) : null,
                        latLng != null ? String.valueOf(latLng.latitude) : null);
                if (response.hasError()) {
                    throw new RuntimeException(response.error.message);
                }

                subscriber.onNext(response.order);
                subscriber.onCompleted();
            }
        }).subscribeOn(Schedulers.io());
    }

    public Observable<ValidatedOrder> validateOrderByAddressId(
            final long groupId, final int addressId) {
        return Observable.create(new Observable.OnSubscribe<ValidatedOrder>() {
            @Override
            public void call(Subscriber<? super ValidatedOrder> subscriber) {
                if (subscriber.isUnsubscribed()) {
                    return;
                }

                ValidateOrderResponse response = mYmxRestService.validateOrderByAddressId(
                        (String) mDataManager.get(CacheKeys.TOKEN), groupId,
                        addressId);
                if (response.hasError()) {
                    throw new RuntimeException(response.error.message);
                }

                subscriber.onNext(response.validatedOrder);
                subscriber.onCompleted();
            }
        }).subscribeOn(Schedulers.io());
    }

    public Observable<ValidatedOrder> validateOrderByAddressId(
            final long groupId, final int addressId, final int deliveryId, final int bonusId) {
        return Observable.create(new Observable.OnSubscribe<ValidatedOrder>() {
            @Override
            public void call(Subscriber<? super ValidatedOrder> subscriber) {
                if (subscriber.isUnsubscribed()) {
                    return;
                }

                ValidateOrderResponse response = mYmxRestService.validateOrderByAddressId(
                        (String) mDataManager.get(CacheKeys.TOKEN), groupId,
                        addressId, deliveryId, bonusId);
                if (response.hasError()) {
                    throw new RuntimeException(response.error.message);
                }

                subscriber.onNext(response.validatedOrder);
                subscriber.onCompleted();
            }
        }).subscribeOn(Schedulers.io());
    }

    public Observable<ValidatedOrder> validateOrderByStoreId(final String goodsJsonString,
                                                             final long groupId, final int storeId) {
        return Observable.create(new Observable.OnSubscribe<ValidatedOrder>() {
            @Override
            public void call(Subscriber<? super ValidatedOrder> subscriber) {
                if (subscriber.isUnsubscribed()) {
                    return;
                }

                ValidateOrderResponse response = mYmxRestService.validateOrderByStoreId(
                        (String) mDataManager.get(CacheKeys.TOKEN), groupId,
                        storeId);
                if (response.hasError()) {
                    throw new RuntimeException(response.error.message);
                }

                subscriber.onNext(response.validatedOrder);
                subscriber.onCompleted();
            }
        }).subscribeOn(Schedulers.io());
    }

    public Observable<Order> createOrder(final String taskId) {
        return Observable.create(new Observable.OnSubscribe<Order>() {
            @Override
            public void call(Subscriber<? super Order> subscriber) {
                if (subscriber.isUnsubscribed()) {
                    return;
                }

                CreateOrderResponse response = mYmxRestService.getOrders(
                        (String) mDataManager.get(CacheKeys.TOKEN),
                        taskId);
                if (response.hasError()) {
                    throw new RuntimeException(response.error.message);
                }

                subscriber.onNext(response.order);
                subscriber.onCompleted();
            }
        }).subscribeOn(Schedulers.io());
    }

    public Observable<ValidatedOrder> validateOrderByStoreId(final String goodsJsonString,
                                                             final long groupId, final int storeId,
                                                             final int deliveryId, final int bonusId) {
        return Observable.create(new Observable.OnSubscribe<ValidatedOrder>() {
            @Override
            public void call(Subscriber<? super ValidatedOrder> subscriber) {
                if (subscriber.isUnsubscribed()) {
                    return;
                }

                ValidateOrderResponse response = mYmxRestService.validateOrderByStoreId(
                        (String) mDataManager.get(CacheKeys.TOKEN), groupId,
                        storeId,
                        deliveryId, bonusId);
                if (response.hasError()) {
                    throw new RuntimeException(response.error.message);
                }

                subscriber.onNext(response.validatedOrder);
                subscriber.onCompleted();
            }
        }).subscribeOn(Schedulers.io());
    }

    public Observable<String> createAsyncOrder(final String orderString, final String longitude,
            final String latitude) {
        return Observable.create(new Observable.OnSubscribe<String>() {
            @Override
            public void call(Subscriber<? super String> subscriber) {
                if (subscriber.isUnsubscribed()) {
                    return;
                }

                AsyncCreateOrderResponse response = mYmxRestService.createAsyncOrder(
                        (String) mDataManager.get(CacheKeys.TOKEN),
                        orderString,
                        mReferFrom,
                        longitude, latitude, Constants.BAIDU_COORD_SYSTEM);
                if (response.hasError()) {
                    throw new RuntimeException(response.error.message);
                }

                subscriber.onNext(response.taskId);
                subscriber.onCompleted();
            }
        }).subscribeOn(Schedulers.io());
    }

    public Observable<Boolean> getVerifyCode(final String phoneNumber) {
        return Observable.create(new Observable.OnSubscribe<Boolean>() {
            @Override
            public void call(Subscriber<? super Boolean> subscriber) {
                if (subscriber.isUnsubscribed()) {
                    return;
                }

                BaseResponse response = mYmxRestService.getVerifyCode(phoneNumber, getUserToken());
                if (response.hasError()) {
                    throw new RuntimeException(response.error.message);
                }

                subscriber.onNext(true);
                subscriber.onCompleted();
            }
        }).subscribeOn(Schedulers.io());
    }

    public Observable<Pair<String, User>> checkUser(final String phoneNumber,
            final String verifyCode) {
        return Observable.create(new Observable.OnSubscribe<Pair<String, User>>() {
            @Override
            public void call(Subscriber<? super Pair<String, User>> subscriber) {
                if (subscriber.isUnsubscribed()) {
                    return;
                }

                CheckUserResponse response = mYmxRestService.checkUser(mReferFrom, phoneNumber,
                        verifyCode, getUserToken());
                if (response.hasError() || response.user == null) {
                    throw new RuntimeException(response.error.message);
                }
                User user = new User();
                user.token = response.token;
                SharedPreferencesHelper.saveUser(user, SharePreferenceKeys.CURRENT_USER);
                subscriber.onNext(Pair.create(response.token, response.user));
                subscriber.onCompleted();
            }
        }).subscribeOn(Schedulers.io());
    }

    public Observable<PullMessage> getPullMessageByToken(final String token) {
        return Observable.create(new Observable.OnSubscribe<PullMessage>() {
            @Override
            public void call(Subscriber<? super PullMessage> subscriber) {
                if (subscriber.isUnsubscribed()) {
                    return;
                }

                PullMessageResponse response = mYmxRestService.getPullMessageByToken(token);
                if (response.hasError()) {
                    throw new RuntimeException(response.error.message);
                }

                subscriber.onNext(response.pullMessage);
                subscriber.onCompleted();
            }
        }).subscribeOn(Schedulers.io());
    }

    public Observable<PullMessage> getPullMessageByLocation(
            final String longitude, final String latitude) {
        return Observable.create(new Observable.OnSubscribe<PullMessage>() {
            @Override
            public void call(Subscriber<? super PullMessage> subscriber) {
                if (subscriber.isUnsubscribed()) {
                    return;
                }

                PullMessageResponse response = mYmxRestService.getPullMessageByLocation
                        (longitude, latitude, Constants.BAIDU_COORD_SYSTEM, getUserToken());
                if (response.hasError()) {
                    throw new RuntimeException(response.error.message);
                }

                subscriber.onNext(response.pullMessage);
                subscriber.onCompleted();
            }
        }).subscribeOn(Schedulers.io());
    }

    public Observable<List<Store>> getStores(
            final String deliveryType, final String regionId, final String longitude,
            final String latitude, final String coordSystem) {
        return Observable.create(new Observable.OnSubscribe<List<Store>>() {
            @Override
            public void call(Subscriber<? super List<Store>> subscriber) {
                if (subscriber.isUnsubscribed()) {
                    return;
                }

                StoresResponse response = mYmxRestService.getStores
                        (deliveryType, regionId, longitude, latitude, coordSystem, getUserToken());
                if (response.hasError()) {
                    throw new RuntimeException(response.error.message);
                }

                subscriber.onNext(response.stores);
                subscriber.onCompleted();
            }
        }).subscribeOn(Schedulers.io());
    }

    public Observable<String> notifyPaySuc(final String paymentId, final String paymethod) {
        return Observable.create(new Observable.OnSubscribe<String>() {
            @Override
            public void call(Subscriber<? super String> subscriber) {
                if (subscriber.isUnsubscribed()) {
                    return;
                }

                BaseResponse response = mYmxRestService.paySuccess(paymentId, paymethod, getUserToken());
                if (response.hasError()) {
                    throw new RuntimeException(response.error.message);
                }

                subscriber.onNext(response.toString());
                subscriber.onCompleted();
            }
        }).subscribeOn(Schedulers.io());
    }

    public Observable<Store> getPickUpStoreGoodsByStoreId(
            final String storeId, final String token, final String longitude, final String latitude) {
        return Observable.create(new Observable.OnSubscribe<Store>() {
            @Override
            public void call(Subscriber<? super Store> subscriber) {
                if (subscriber.isUnsubscribed()) {
                    return;
                }

                GetStoreResponse response = mYmxRestService.getStoreByStoreId(storeId
                        , 0, longitude, latitude, Constants.BAIDU_COORD_SYSTEM, token);
                if (response.hasError()) {
                    throw new RuntimeException(response.error.message);
                }

                subscriber.onNext(response.store);
                subscriber.onCompleted();
            }
        }).subscribeOn(Schedulers.io());
    }

    public String getEndPoint() {
        switch (BuildConfig.SERVER_ENVIRONMENT) {
            case Release:
                return RELEASE_ENDPOINT;
            case Staging:
                return STAGING_ENDPOINT;
            case Test:
                return TEST_ENDPOINT;
            case Mock:
                return Constants.MOCK_END_POINT;
            case RD:
                return RD_ENDPOINT;
            default:
                return RELEASE_ENDPOINT;
        }
    }

    /**
     * 更新用户信息
     */
    public Observable<User> updateUserInfo(final User user) {
        if (user.careerMap != null) {
            for (KV careerKv : user.careerMap) {
                if (careerKv.value.equals(user.career)) {
                    user.careerCode = careerKv.key;
                    break;
                }
            }
        }

        return Observable.create(new Observable.OnSubscribe<User>() {
            @Override
            public void call(Subscriber<? super User> subscriber) {
                if (subscriber.isUnsubscribed()) {
                    return;
                }
                UpdateUserInfoResponse response = mYmxRestService.updateUserInfo(
                        mDataManager.<String>get(CacheKeys.TOKEN),
                        user.nickName, user.sex, user.age, user.careerCode);
                if (response.hasError() || response.user == null) {
                    throw new RuntimeException(response.error.message);
                }
                User user = response.user;
                user.token = mDataManager.<String> get(CacheKeys.TOKEN);
                SharedPreferencesHelper.saveUser(response.user, SharePreferenceKeys.CURRENT_USER);
                subscriber.onNext(response.user);
                subscriber.onCompleted();
            }
        }).subscribeOn(Schedulers.io());
    }

    /**
     * 获取广告URL
     */
    public Observable<AdvertInfo> getAdInfo(final String region, final String channel) {
        return Observable.create(new Observable.OnSubscribe<AdvertInfo>() {
            @Override
            public void call(Subscriber<? super AdvertInfo> subscriber) {
                if (subscriber.isUnsubscribed()) {
                    return;
                }
                GetAdvertInfoResponse response = mYmxRestService.getAdInfo(
                        region, channel, getUserToken());
                if (response.hasError()) {
                    throw new RuntimeException(response.error.message);
                }
                AdvertInfo advertInfo = response.data;
                if (response.res == 200 && advertInfo != null) {
                    subscriber.onNext(advertInfo);
                } else {
                    advertInfo = new AdvertInfo();
                    advertInfo.url = "";
                    mDataManager.saveAdInfo(advertInfo);
                }
                subscriber.onCompleted();
            }
        }).subscribeOn(Schedulers.io());
    }

    /**
     * 获取商品详情信息
     */
    /*
     * public Observable<GoodDetail> getGoodDetail(final String id) { if
     * (TextUtils.isEmpty(id)) { return null; } return Observable.create(new
     * Observable.OnSubscribe<GoodDetail>() {
     * @Override public void call(Subscriber<? super GoodDetail> subscriber) {
     * if (subscriber.isUnsubscribed()) { return; } GoodDetailResponse response
     * = mYmxRestService.getGoodDetailInfo(id); if (response.hasError()) { throw
     * new RuntimeException(response.error.message); }
     * subscriber.onNext(response.goodDetail); subscriber.onCompleted(); }
     * }).subscribeOn(Schedulers.io()); }
     */

    public YmxRestService getYmxRestService() {
        return mYmxRestService;
    }

    public String getUserToken(){
        User user = SharedPreferencesHelper.getUser(SharePreferenceKeys.CURRENT_USER);
        if(user != null && user.token != null){
            return user.token;
        }
        return "";
    }
}
